package cn.codenest.listener;

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;

import java.util.Collection;
import java.util.HashMap;
import java.util.Map;

/**
 * @author ：Hyman
 * @date ：Created in 2021/7/9 18:31
 * @description： 普通的再均衡监听器，再均衡时进行相应处理。
 * @modified By：
 * @version: $
 */
public class HandleRebalanceListener implements ConsumerRebalanceListener {

    public HandleRebalanceListener(Map<TopicPartition, OffsetAndMetadata> currentOffsets, Consumer consumer) {
        this.currentOffsets = currentOffsets;
        this.consumer = consumer;
    }

    private Map<TopicPartition, OffsetAndMetadata> currentOffsets = new HashMap<>(128);
    private Consumer consumer;

    /**
     * create by: Hyman
     * description: 在再均衡开始之前，消费者停止读取消息之后被调用,如果在这里提交
     * 偏移量，那么下一个接管分区的消费者就知道从哪里开始读了
     * create time: 2021/7/9
     */
    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> collection) {
        System.out.println("Lost partitions in rebalance.Committing current offsets:"
                + currentOffsets);
        consumer.commitSync(currentOffsets);
    }

    /**
     * create by: Hyman
     * description: 会在重新分配分区之后和消费者开始读取消息之前被调用
     * create time: 2021/7/9
     */
    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> collection) {

    }
}
